package transform;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.junit.Test;

/**
 * @author dinghao
 * @create 2021-09-28 18:55
 * @message
 */
public class Flink_transform_join {
    //TODO join算子测试,分别往两个socket网络端口中输入单词然后回车，每10s内有相同单词的会关联成   word-word 格式
    @Test
    public void test1() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        DataStreamSource<String> node190 = env.socketTextStream("node190", 9999);
        DataStreamSource<String> node193 = env.socketTextStream("node193", 9999);

        node190.join(node193)
                .where(value -> value)
                .equalTo(value -> value)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .apply((JoinFunction<String, String, String>) (first, second) -> first+"-"+second)
                .print();
        env.execute();
    }
}
